package com.amos.mqserver.service;

import com.alibaba.fastjson.JSONObject;
import com.amos.mqserver.pojo.Constants;
import com.amos.mqserver.pojo.ExchangeType;
import com.amos.mqserver.pojo.MqMessage;
import com.amos.mqserver.util.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author liangh on 2018/9/12 0012.
 */
@Service
@DependsOn("rabbitT")
public class MsgSendClient implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

    Logger logger = LoggerFactory.getLogger(MsgSendClient.class);

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private JdbcTemplate jdbcTemplate;


    @Autowired
    private RabbitAdmin rabbitAdmin;

    private Integer messagePriority = 1;

    /**
     * 初始化rabbitTemplate参数
     */
    @PostConstruct
    private void init() {
        logger.info("rabbitTemplate : init : {}", rabbitTemplate);
        this.rabbitTemplate.setConfirmCallback(this);
        this.rabbitTemplate.setReturnCallback(this);
        this.rabbitTemplate.setMandatory(true);
    }

    /**
     * 发送消息方法
     *
     * @param mqMessage
     * @throws Exception
     */
    public void sendMessage(MqMessage mqMessage) throws Exception {
        logger.info("------sendMessage start-------");
        logger.info("sendMessage : messageId: {}, messageBody: {}, exchangeType: {}, exchangeName: {}, queueName: {}, routeKey: {}, msgOrder: {}", mqMessage.getMessageId(),
                mqMessage.getMessageBody(), mqMessage.getExchangeType(), mqMessage.getExchangeName(), mqMessage.getQueueName(), mqMessage.getRouteKey(), mqMessage.getMsgOrder());
        //验证数据合理性
        validateMessage(mqMessage);
        //动态绑定处理
        if (StringUtils.isEmpty(mqMessage.getQueueName())) {
            mqMessage.setQueueName("justSend");
        }
        dynamicBinder(mqMessage);
        //初始化消息id
        CorrelationData correlationData = new CorrelationData();
        if (StringUtils.isEmpty(mqMessage.getMessageId())) {
            mqMessage.setMessageId(UUID.randomUUID().toString().replaceAll("-", ""));
        }
        correlationData.setId(mqMessage.getMessageId());
        //保存消息记录到表中
        saveMessageTraceIntoTable(mqMessage);
        //有序消息处理
        Map map = new HashMap();
        map.put("messageId", mqMessage.getMessageId());
        map.put("messageBody", mqMessage.getMessageBody());
        if (!StringUtils.isEmpty(mqMessage.getMsgOrder())) {
            map.put("msgOrder", mqMessage.getMsgOrder());
        }
        JSONObject jsonObject = new JSONObject(map);
        String messageBody = jsonObject.toString();
        //分是否有消息优先级两种情况来发送消息
        if (StringUtils.isEmpty(mqMessage.getMessagePriority())) {
            this.rabbitTemplate.convertAndSend(mqMessage.getExchangeName(), mqMessage.getRouteKey(), messageBody, correlationData);
        } else {
            messagePriority = mqMessage.getMessagePriority();
            this.rabbitTemplate.convertAndSend(mqMessage.getExchangeName(), mqMessage.getRouteKey(), messageBody, new MessagePostProcessor() {
                @Override
                public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
                    message.getMessageProperties().setPriority(messagePriority);
                    logger.info("messagePriority : {}", messagePriority);
                    return message;
                }
            }, correlationData);
        }
        logger.info("------sendMessage end-------");
    }

    /**
     * 验证消息体内容的是否符合要求
     *
     * @param mqMessage
     * @throws Exception
     */
    private void validateMessage(MqMessage mqMessage) throws Exception {
        logger.info("------sendMessage validateMessage start-------");
        boolean flag = false;
        if (StringUtils.isEmpty(mqMessage.getExchangeType())) {
            throw new RuntimeException("交换机类型不能为空");
        }
        for (ExchangeType exchangeType : ExchangeType.values()) {
            if (exchangeType.getExchangeType().equals(mqMessage.getExchangeType())) {
                flag = true;
            }
        }
        if (!flag) {
            throw new RuntimeException("交换机类型不正确");
        }
        if (StringUtils.isEmpty(mqMessage.getExchangeName())) {
            throw new RuntimeException("交换机名称不能为空");
        }
        if (StringUtils.isEmpty(mqMessage.getMessageBody())) {
            throw new RuntimeException("消息内容不能为空");
        }
        if (StringUtils.isEmpty(mqMessage.getRouteKey())) {
            if (!mqMessage.getExchangeType().equals(ExchangeType.FANOUT_EXCHANGE.getExchangeType())) {
                throw new RuntimeException("非fanout类型的交换机，路由键不能为空");
            }
        }
        logger.info("------sendMessage validateMessage end-------");
    }

    /**
     * 动态绑定队列
     *
     * @param mqMessage
     * @throws Exception
     */
    public void dynamicBinder(MqMessage mqMessage) throws Exception {
        logger.info("------sendMessage dynamicBinder start-------");
        if (StringUtils.isEmpty(mqMessage.getMessageBody())) {
            mqMessage.setMessageBody("no thing");
        }
        validateMessage(mqMessage);
        if (!mqMessage.getQueueName().equals("justSend")) {
            if (!StringUtils.isEmpty(mqMessage.getQueueName())) {
                Binding binding = null;
                Queue queue;
                if (!StringUtils.isEmpty(mqMessage.getQueuePriority())) {
                    Map<String, Object> args = new HashMap<>(1);
                    args.put("x-max-priority", mqMessage.getQueuePriority());
                    queue = new Queue(mqMessage.getQueueName(), true, false, false, args);
                } else {
                    queue = new Queue(mqMessage.getQueueName());
                }
                rabbitAdmin.declareQueue(queue);
                if (mqMessage.getExchangeType().equals(ExchangeType.DIRECT_EXCHANGE.getExchangeType())) {
                    DirectExchange directExchange = new DirectExchange(mqMessage.getExchangeName(), true, false);
                    binding = BindingBuilder.bind(new Queue(mqMessage.getQueueName())).to(directExchange).with(mqMessage.getRouteKey());
                }
                if (mqMessage.getExchangeType().equals(ExchangeType.TOPIC_EXCHANGE.getExchangeType())) {
                    TopicExchange topicExchange = new TopicExchange(mqMessage.getExchangeName(), true, false);
                    binding = BindingBuilder.bind(new Queue(mqMessage.getQueueName())).to(topicExchange).with(mqMessage.getRouteKey());
                }
                if (mqMessage.getExchangeType().equals(ExchangeType.FANOUT_EXCHANGE.getExchangeType())) {
                    FanoutExchange fanoutExchange = new FanoutExchange(mqMessage.getExchangeName(), true, false);
                    binding = BindingBuilder.bind(new Queue(mqMessage.getQueueName())).to(fanoutExchange);
                }
                rabbitAdmin.declareBinding(binding);
                logger.info("dynamicBinder : exchangeName : {}, queueName : {}, routeKey : {}", mqMessage.getExchangeName(), mqMessage.getQueueName(), mqMessage.getRouteKey());
            } else {
                throw new RuntimeException("队列名称不能为空");
            }
        }
        logger.info("------sendMessage dynamicBinder end-------");
    }

    /**
     * 将消息内容保存到消息记录表中
     *
     * @param mqMessage
     */
    private void saveMessageTraceIntoTable(MqMessage mqMessage) {
        logger.info("------sendMessage saveMessageTraceIntoTable start-------");
        //查询是否存在该消息记录
        String querySql = "select * from rabbitmq_message_trace where message_id = '" + mqMessage.getMessageId() + "'";
        List<Map<String, Object>> msgTraceList = jdbcTemplate.queryForList(querySql);
        if (msgTraceList.isEmpty()) {
            //不存在该消息记录，即可直接插入数据
            Date date = new Date();
            Integer priority = StringUtils.isEmpty(mqMessage.getMessagePriority()) ? 0 : mqMessage.getMessagePriority();
            String insertSql = "insert into rabbitmq_message_trace(message_id, exchange_type, exchange_name, route_key, message_priority, message_body, message_status, try_num, cretime)" +
                    " values('" + mqMessage.getMessageId() + "','" + mqMessage.getExchangeType() + "','" + mqMessage.getExchangeName() + "','" + mqMessage.getRouteKey() + "','" +
                    priority + "','" + mqMessage.getMessageBody() + "','" + Constants.MESSAGE_UN_RECEIVE_CONFIRM + "',0" + ",DATE_FORMAT('" + DateUtil.formatDate(date) + "','%Y%m%d%H%i%s'))";
            logger.info("insertSql : {}", insertSql);
            jdbcTemplate.execute(insertSql);
            logger.info("insert into rabbitmq_message_trace(message_id: {}, exchange_type: {}, exchange_name: {}, route_key: {}, message_priority: {}, message_body: {}, message_status: {}, try_num: {}, cretime: {})" +
                    mqMessage.getMessageId(), mqMessage.getExchangeType(), mqMessage.getExchangeName(), mqMessage.getRouteKey(), mqMessage.getMessagePriority(), mqMessage.getMessageBody(), Constants.MESSAGE_UN_RECEIVE_CONFIRM, 0, date);
        }
        logger.info("------sendMessage saveMessageTraceIntoTable end-------");
    }

    /**
     * 回调方法confirm
     * 如果消息没有到exchange,则confirm回调,ack=false
     * 如果消息到达exchange,则confirm回调,ack=true
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        logger.info("------sendMessage confirm start-------");
        logger.info("correlationData Id: {}, ack: {}, cause: {}", correlationData.getId(), ack, cause);
        try {
            String querySql = "select * from rabbitmq_message_trace where message_id = '" + correlationData.getId() + "'";
            List<Map<String, Object>> msgTraceList = jdbcTemplate.queryForList(querySql);
            if (!msgTraceList.get(0).get("message_status").equals(Constants.MESSAGE_SEND_ERROR)) {
                if (ack) {
                    //mq接收到消息，正常发送回馈确认，修改消息记录表里面消息状态为“发送成功” 4
                    String successSql = "update rabbitmq_message_trace set message_status ='" + Constants.MESSAGE_SEND_SUCCESS + "', modtime = '"
                            + new SimpleDateFormat("yyyy-MM-dd  HH:mm:ss").format(new Date()) + "' where message_id = '" + correlationData.getId() + "'";
                    jdbcTemplate.update(successSql);
                } else {
                    if (!msgTraceList.get(0).get("message_status").equals(Constants.MESSAGE_SEND_FAIL_COMPENSATE)) {
                        String updateSql = "update rabbitmq_message_trace set message_status = '" + Constants.MESSAGE_SEND_ERROR + "', modtime = '"
                                + new SimpleDateFormat("yyyy-MM-dd  HH:mm:ss").format(new Date()) + "', send_error = '" + cause.replace("\'", "\"") + "' where message_id = '" + correlationData.getId() + "'";
                        logger.info("updateSql : {}", updateSql);
                        jdbcTemplate.execute(updateSql);
                    }
                    throw new RuntimeException("可能由于交换机名称错误或者网络异常，导致发送失败，请检查交换机名称正确性后重新发送消息，该消息的id是" + correlationData.getId());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        logger.info("------sendMessage confirm end-------");
    }

    /**
     * 回调方法returnedMessage
     * exchange到queue成功,则不回调return
     * exchange到queue失败,则回调return(需设置mandatory=true,否则不回回调,消息就丢了)
     */
    @Override
    public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.info("------sendMessage returnedMessage start-------");
        logger.info("returnedMessage : messageBody : {}, replyCode : {}, replyText : {}, exchange : {}, routingKey : {}", new String(message.getBody()), replyCode, replyText, exchange, routingKey);
        String updateSql = "update rabbitmq_message_trace set message_status = '" + Constants.MESSAGE_SEND_ERROR + "', send_error = '" + replyText + "', modtime = '" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) +
                "' where exchange_name = '" + exchange + "' and route_key = '" + routingKey + "' and message_body = '" + new String(message.getBody()) + "'";
        jdbcTemplate.execute(updateSql);
        logger.info("------sendMessage returnedMessage end-------");
        throw new RuntimeException("可能由于路由键错误或者网络异常，导致发送失败，请检查路由键正确性后重新发送消息，该消息的内容是：" + new String(message.getBody()));
    }

}
